package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo9ReduceByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("reduceByKey算子演示")
    val context = new SparkContext(conf)
    //====================================================
    val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")

    val splitRDD: RDD[Array[String]] = studentRDD.map((s: String) => s.split(","))

    //求每个班级的人数
    val clazzKVRDD: RDD[(String, Int)] = splitRDD.map {
      case Array(_, _, _, _, clazz: String) => (clazz, 1)
    }

    /**
     * 利用groupByKey实现
     */
    //    val kvRDD: RDD[(String, Iterable[Int])] = clazzKVRDD.groupByKey()
    //    val clazzAvgAgeRDD: RDD[(String, Double)] = kvRDD.map {
    //      case (clazz: String, n: Iterable[Int]) =>
    //        (clazz, n.sum)
    //    }
    //    clazzAvgAgeRDD.foreach(println)

    /**
     * 利用reduceByKey实现：按照键key对value值直接进行聚合，需要传入聚合的方式
     * reduceByKey算子也是只有kv类型的RDD才能调用
     *
     *
     */
    val countRDD: RDD[(String, Int)] = clazzKVRDD.reduceByKey((x: Int, y: Int) => x + y)
    countRDD.foreach(println)



//    clazzKVRDD.groupByKey()
//      .map(kv=>(kv._1,kv._2.sum))
//      .foreach(println)

    while (true){

    }

    /**
     *  reduceByKey与groupByKey的区别
     *  1、reduceByKey比groupByKey在map端多了一个预聚合的操作，预聚合之后的shuffle数据量肯定是要少很多的，性能上比groupByKey要好
     *  2、从灵活角度来看，reduceByKey并没有groupByKey灵活
     *   比如reduceByKey无法做方差，groupByKey后续可以完成
     *
     */



  }
}
